package com.ycz.publish;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmListener;
import com.rabbitmq.client.Connection;
import com.ycz.config.RabbitConnectionUtil;
import com.ycz.consumer.ConsumerTest;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.Date;

/**
 * 消息持久化
 * Rabbit队列和交换器有一个不可告人的秘密，就是默认情况下重启服务器会导致消息丢失，那么怎么保证Rabbit在重启的时候不丢失呢？答案就是消息持久化。
 * <p>
 * 当你把消息发送到Rabbit服务器的时候，你需要选择你是否要进行持久化，但这并不能保证Rabbit能从崩溃中恢复，想要Rabbit消息能恢复必须满足3个条件：
 * <p>
 * 投递消息的时候durable设置为true，消息持久化，代码：channel.queueDeclare(x, true, false, false, null)，参数2设置为true持久化；
 * 设置投递模式deliveryMode设置为2（持久），代码：channel.basicPublish(x, x, MessageProperties.PERSISTENT_TEXT_PLAIN,x)，参数3设置为存储纯文本到磁盘；
 * 消息已经到达持久化交换器上；
 * 消息已经到达持久化的队列；
 * 持久化工作原理
 * <p>
 * Rabbit会将你的持久化消息写入磁盘上的持久化日志文件，等消息被消费之后，Rabbit会把这条消息标识为等待垃圾回收。
 *
 * @author ycz
 * @date 2021/6/11
 *
 */
@Component
public class PublishTest {

    static String queueName = "hello";
    String exchangeName = "helloExchange";
    String topicExchange = "topicExchange";

    /**
     * fanout交换器——发布/订阅模式
     */
    public void publisherByFanout() {
        Connection conn = RabbitConnectionUtil.getRabbitConnection();
        if (conn != null) {
            try {
                // 创建通道
                Channel channel = conn.createChannel();
                // 声明队列【参数说明：参数一：队列名称，参数二：是否持久化；参数三：是否独占模式；参数四：消费者断开连接时是否删除队列；参数五：消息其他参数】
                // channel.queueDeclare(queueName,true,false,false,null);
                // 声明fanout交换器
                channel.exchangeDeclare(exchangeName, "fanout");
                String content = String.format("当前时间：%s", new Date().getTime());
                // 发送内容【参数说明：参数一：交换机名称；参数二：队列名称，参数三：消息的其他属性-routing headers，此属性为MessageProperties.PERSISTENT_TEXT_PLAIN用于设置纯文本消息存储到硬盘；参数四：消息主体】
                // TODO 没有指定队列名
                channel.basicPublish(exchangeName, "", null, content.getBytes("UTF-8"));
                System.out.println("已发送消息：" + content);
                // 关闭连接
                channel.close();
                conn.close();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * direct 交换器
     */
    public void publisher() {
        Connection conn = RabbitConnectionUtil.getRabbitConnection();
        if (conn != null) {
            try {
                // 创建通道
                Channel channel = conn.createChannel();
                // 声明队列【参数说明：参数一：队列名称，参数二：是否持久化；参数三：是否独占模式；参数四：消费者断开连接时是否删除队列；参数五：消息其他参数】
                channel.queueDeclare(queueName, true, false, false, null);
                String content = String.format("当前时间：%s", new Date().getTime());
                // 发送内容【参数说明：参数一：交换机名称；参数二：队列名称，参数三：消息的其他属性-routing headers，此属性为MessageProperties.PERSISTENT_TEXT_PLAIN用于设置纯文本消息存储到硬盘；参数四：消息主体】
                channel.basicPublish("", queueName, null, content.getBytes("UTF-8"));
                System.out.println("已发送消息：" + content);
                // 关闭连接
                channel.close();
                conn.close();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * topic交换器 —— 匹配订阅模式
     * 可以更灵活的匹配自己想要订阅的信息，这个时候routingKey路由键就排上用场了，使用路由键进行消息（规则）匹配
     */
    public void publishByTopic() {
        String routingKey = "com.mq.rabbit.error";
        Connection conn = RabbitConnectionUtil.getRabbitConnection();
        if (conn != null) {
            try {
                // 创建通道
                Channel channel = conn.createChannel();
                // 声明队列【参数说明：参数一：队列名称，参数二：是否持久化；参数三：是否独占模式；参数四：消费者断开连接时是否删除队列；参数五：消息其他参数】
                // 声明topic交换器
                channel.exchangeDeclare(topicExchange, "topic");
                String content = String.format("当前时间：%s", new Date().getTime());
                // 发送内容【参数说明：参数一：交换机名称；参数二：队列名称，参数三：消息的其他属性-routing headers，此属性为MessageProperties.PERSISTENT_TEXT_PLAIN用于设置纯文本消息存储到硬盘；参数四：消息主体】
                channel.basicPublish(topicExchange, routingKey, null, content.getBytes("UTF-8"));
                System.out.println("publishByTopic 已发送消息：" + content);
                // 关闭连接
                channel.close();
                conn.close();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    public void publishWithTransaction() {
        Channel channel = null;
        try {
            Connection conn = RabbitConnectionUtil.getRabbitConnection();
            // 创建信道
            channel = conn.createChannel();
            // 声明队列【参数说明：参数一：队列名称，参数二：是否持久化；参数三：是否独占模式；参数四：消费者断开连接时是否删除队列；参数五：消息其他参数】
            channel.queueDeclare(queueName, true, false, false, null);
            String content = String.format("当前时间：%s", new Date().getTime());

            channel.txSelect();
            System.out.println("rabbit 已开启事务");

            // 发送内容【参数说明：参数一：交换机名称；参数二：队列名称，参数三：消息的其他属性-routing headers，此属性为MessageProperties.PERSISTENT_TEXT_PLAIN用于设置纯文本消息存储到硬盘；参数四：消息主体】
            channel.basicPublish("", queueName, null, content.getBytes("UTF-8"));
            System.out.println("已发送消息：" + content);
            channel.txCommit();
            System.out.println("rabbit 已提交事务");

            // 关闭连接
            channel.close();
            conn.close();
        } catch (Exception e) {
            try {
                channel.txRollback();
                System.out.println("rabbit 事务回滚");
            } catch (IOException ioException) {
            }
        }
    }

    /**
     * 普通发送方confirm模式
     * https://www.cnblogs.com/vipstone/p/9350075.html
     * Confirm模式要比事务快10倍左右
     * 非事务模式的性能是事务模式的性能高149倍
     */
    public void publishWithWaitForConfirms() {
        try {
            Connection conn = RabbitConnectionUtil.getRabbitConnection();
            // 创建信道
            Channel channel = conn.createChannel();
            // 声明队列
            channel.queueDeclare(queueName, true, false, false, null);
            // 开启发送方确认模式
            channel.confirmSelect();

            String content = String.format("当前时间：%s", new Date().getTime());
            // 发送内容【参数说明：参数一：交换机名称；参数二：队列名称，参数三：消息的其他属性-routing headers，此属性为MessageProperties.PERSISTENT_TEXT_PLAIN用于设置纯文本消息存储到硬盘；参数四：消息主体】
            channel.basicPublish("", queueName, null, content.getBytes("UTF-8"));
            System.out.println("publishWithWaitForConfirms 已发送消息：" + content);
            if (channel.waitForConfirms()) {
                System.out.println("publishWithWaitForConfirms 已发送消息成功：" + content);
            }
            // 关闭连接
            channel.close();
            conn.close();
        } catch (Exception e) {
        }
    }

    /**
     * 批量Confirm模式
     */
    public void publishWaitForConfirmsOrDie() {
        try {
            Connection conn = RabbitConnectionUtil.getRabbitConnection();
            // 创建信道
            Channel channel = conn.createChannel();
            // 声明队列
            channel.queueDeclare(queueName, true, false, false, null);
            // 开启发送方确认模式
            channel.confirmSelect();

            for (int i = 0; i < 100; i++) {
                String message = String.format("i===", i, "时间 => %s", new Date().getTime());
                channel.basicPublish("", queueName, null, message.getBytes("UTF-8"));
            }
            //  //直到所有信息都发布，只要有一个未确认就会IOException
            channel.waitForConfirmsOrDie();
            System.out.println("waitForConfirmsOrDie 全部执行完成");
            // 关闭连接
            channel.close();
            conn.close();
        } catch (Exception e) {
        }
    }

    /**
     * 异步Confirm模式
     */
    public void publishWaitForConfirmsListener() {
        try {
            Connection conn = RabbitConnectionUtil.getRabbitConnection();
            // 创建信道
            Channel channel = conn.createChannel();
            // 声明队列
            channel.queueDeclare(queueName, true, false, false, null);
            // 开启发送方确认模式
            channel.confirmSelect();

            for (int i = 0; i < 100; i++) {
                String message = String.format("i===", i, "时间 => %s", new Date().getTime());
                channel.basicPublish("", queueName, null, message.getBytes("UTF-8"));
            }
            // 异步监听确认和未确认的消息
            channel.addConfirmListener(new ConfirmListener() {
                @Override
                public void handleAck(long deliveryTag, boolean multiple) throws IOException {
                    System.out.println("未确认消息，标识：" + deliveryTag);
                }

                @Override
                public void handleNack(long deliveryTag, boolean multiple) throws IOException {
                    System.out.println(String.format("已确认消息，标识：%d，多个消息：%b", deliveryTag, multiple));
                }
            });
            // 关闭连接
            channel.close();
            conn.close();
        } catch (Exception e) {
        }
    }


    public static void main(String[] args) {
        PublishTest publishTest = new PublishTest();
        publishTest.publishWithTransaction();
//        ConsumerTest consumerTest = new ConsumerTest();
//        publishTest.publisher();
//        consumerTest.consumer();
    }

}
